package com.yuyu.excel.service.excel;

import cn.hutool.core.util.StrUtil;
import com.alibaba.ttl.TransmittableThreadLocal;
import com.alibaba.ttl.TtlRunnable;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.yuyu.excel.utils.ExceptionUtil;
import com.yuyu.excel.utils.TenantContextHolder;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadFactory;

import static com.baomidou.mybatisplus.extension.toolkit.Db.saveBatch;

@Component
@Slf4j
public class DataRecordDisruptor<T> {

    // Disruptor环形缓冲区的大小，可根据需要调整
    private static final int BUFFER_SIZE = 1024 * 1024;
    // 批量数
    private static final int BATCH_SIZE = 100000;
    private final Disruptor<DataRecordEvent<T>> disruptor;
    private final RingBuffer<DataRecordEvent<T>> ringBuffer;

    @SuppressWarnings("unchecked")
    public DataRecordDisruptor() {
        // 事件工厂
        EventFactory<DataRecordEvent<T>> factory = new MyEventFactory<>();
//        ExecutorService ttlExecutorService = TtlExecutors.getTtlExecutorService(Executors.newFixedThreadPool(10));

        // 初始化Disruptor，使用多生产者模式和阻塞等待策略
        disruptor = new Disruptor<>(factory, BUFFER_SIZE, new ThreadFactory() {
            private final ThreadFactory defaultFactory = Thread::new;

            @Override
            public Thread newThread(Runnable r) {
                return defaultFactory.newThread(TtlRunnable.get(r));
            }
        }, ProducerType.MULTI, new SleepingWaitStrategy());
//        disruptor = new Disruptor<>(factory, BUFFER_SIZE, ttlExecutorService, ProducerType.MULTI, new SleepingWaitStrategy());

        // 注册Disruptor事件处理器
        DataRecordEventHandle<T> eventHandler = new DataRecordEventHandle<>();
        disruptor.handleEventsWith(eventHandler);
        log.info("Registered event handler: {}", eventHandler);

        // 启动 Disruptor
        ringBuffer = disruptor.start();
        log.info("Disruptor started with buffer size: {}", BUFFER_SIZE);
    }

    public void publish(List<T> dataRecords) {
        for (T record : dataRecords) {
            long sequence = ringBuffer.next(); // 获取下一个序列
            try {
                DataRecordEvent<T> event = ringBuffer.get(sequence);
                event.setRecord(record); // 将记录设置到事件中
                log.debug("Published record to sequence {}: {}", sequence, record);
            } finally {
                ringBuffer.publish(sequence); // 发布序列
            }
        }
    }

    public static class MyEventFactory<T> implements EventFactory<DataRecordEvent<T>> {
        @Override
        public DataRecordEvent<T> newInstance() {
            return new DataRecordEvent<>();
        }
    }

    @Data
    public static class DataRecordEvent<T> {
        private T record;
    }

    public static class DataRecordEventHandle<T> implements EventHandler<DataRecordEvent<T>> {
        private final List<T> batch = new ArrayList<>();

        @Override
        public void onEvent(DataRecordEvent<T> event, long sequence, boolean endOfBatch) {
            String tenantId = TenantContextHolder.getTenantId();
            if (StrUtil.isNotBlank(tenantId)) {
                System.out.println("租户id: " + tenantId);
            } else {
                System.out.println("租户id获取为null");
            }
            try {
                T record = event.getRecord();
                if (record != null) {
                    batch.add(record);
                    log.debug("Added record to batch: {}", record);
                }

                if (batch.size() >= BATCH_SIZE || endOfBatch) {
                    saveBatch(batch, BATCH_SIZE); // 保存批次
                    batch.clear();
                }
            } catch (Exception e) {
                log.error("Excel导入时出现异常: {}", ExceptionUtil.getExceptionMessage(e));
            } finally {
                event.setRecord(null); // 清理事件
            }
        }


    }
}